package day09

import org.apache.spark.{HashPartitioner, SparkConf}
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

object  LoadTopicData {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root")
    val conf = new SparkConf().setAppName("zz").setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(2))

    // 因为要进行批次累加，所以要先设置checkpoint。
    // 注意checkpoint产生的数据在程序结束后也存在磁盘中。因此要清除的话一般是手动清除，或者使用hdfs代码清除
    ssc.checkpoint("hdfs://master:9000/out-2020-7-15")

    // 设置请求kafka的配置信息，这里我用Array(zkQuorum, group, topics, numThread)声明变量。
    // zk列表、group组、topics（如果是多个topic,用空格或者逗号隔开，但是后面自己要分割）、用于获取topic的线程数（其实就是获取topic的consumer数量）
    /*这样写死配置信息会有toString的坑
    val Array(zkQuorum, group, topics, numThread) =
    Array("master:2181,slave1:2181,slave2:2181","group01","test01",1) //这里topic我只写了一个，我的分区也设为固定值1

    // 将topic封装到map里，分割。这里有个坑，Array封装的数据本身就是字符串了，可是这里的必须还是得先toString后才能为字符串。
    topics.toString.split(",").map((_,numThread.toString.toInt))  报错
    */
    // 对接kafka
    // 设置请求kafka的配置信息，这里我用Array(zkQuorum, group, topics, numThread) = args声明变量。这里这样写仅仅是为了学习方便
    // 这样写就默认全部是字符串，而且本身这里封装的数据就是要通过arg传递过来的。
    // 实际开发中，这个Array的配置信息并不是这样来的。而是将这些配置信息放到某一个文件中，然后我们通过别的方式读取到该文件中相应的配置信息然后赋值给对应的(zkQuorum, group, topics, numThread)
    val Array(zkQuorum, group, topics, numThread) = args
    // 将topic封装到map里
    val topicMap: Map[String, Int] = topics.split(",").map((_,numThread.toInt)).toMap
    /**
      * Streaming获取kafka数据有两种方式：
      *   一种是Receiver方式获取数据（高级）：用kafka工具类KafkaUtils的createStream方法获取数据。
      *       该方法必须传递一个存储级别的参数：因为该方法获取kafka数据是先将数据放到缓存里，然后用Excutor来进行分析，这是Streaming提供的方法文不是Kafka的提供的
      *   另一种Direct方式（低级）：不用缓冲到某个地方再拿，而是直连方式，拿数据就消费。
      * 这里用Receiver方式获取kafka数据。
      */
    val dstream: ReceiverInputDStream[(String, String)] =
      KafkaUtils.createStream(ssc, zkQuorum, group, topicMap, StorageLevel.MEMORY_AND_DISK)

    // 因为获取到kafka的数据中是 key,value形式的，其中key为offset，在实际的统计中不需要这个key，可以过过滤掉
    val lines: DStream[String] = dstream.map(_._2)
    //开始统计
    val tups: DStream[(String, Int)] = lines.flatMap(_.split(" ")).map((_,1))
    val res = tups.updateStateByKey(func, new HashPartitioner(ssc.sparkContext.defaultMinPartitions), true)

    res.print()
    ssc.start()
    ssc.awaitTermination()


  }

  val func = (it: Iterator[(String, Seq[Int], Option[Int])]) => {
    //注意，语法格式：map里面要用模式匹配的话，map后面不能接小括号，只能用大括号。
    it.map {
      case (x,y,z) => {
        (x, y.sum + z.getOrElse(0))
      }
    }
  }

}
/*
* IDEA本地运行直接加arg参数，的配置步骤：
*   IDEA上方导航栏有个本类文件名的按钮，点Edit Configurations（如果里面没有本类文件，就点+号Application，然后用在输入name后在Main class找到要运行的class类），
*   然后在Program arguments后输入要传入给程序args的参数，注意参数与参数之间用空格隔开。
*     master:2181,slave1:2181,slave2:2181 group01 test1 1
*   运行本代码前要在虚拟机kafka里提前创建好消息队列test01，运行后在虚拟机里用kafka生产者输入数据
* */

